package com.sun.kafka.producer;

import com.alibaba.fastjson2.JSONObject;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.util.Date;

/**
 * 生产者
 */
@Slf4j
@Component
@RequiredArgsConstructor(onConstructor = @__(@Autowired))
public class KafkaProducer {

    private final KafkaTemplate kafkaTemplate;

    @Value("${monitor.kafka.topics:topic-test}")
    private String topics;

    private static final Gson gson = new GsonBuilder().create();

    public void send(Integer num) {
        for (int i = 0; i < num; i++) {
            for (String topic : topics.split(",")) {
                // kafka不同topic的异构数据
                JSONObject message = new JSONObject();
                if (StringUtils.equals("batch1", topic)) {
                    message.put("name", "metric-" + topic);
                    message.put("ts", new Date());
                    message.put("labels", "{'tagKey':'" + i + "'}");
                    message.put("fields", "{'valueKey':'" + i + "'}");
                    log.info("发送消息：topic = {}, message = {}", topic, message);
                }
                if (StringUtils.equals("batch2", topic)) {
                    message.put("name2", "metric-" + topic);
                    message.put("ts2", new Date());
                    message.put("labels2", "{'tagKey':'" + i + "'}");
                    message.put("fields2", "{'valueKey':'" + i + "'}");
                    log.info("发送消息：topic = {}, message = {}", topic, message);
                }
                kafkaTemplate.send(topic, gson.toJson(message));
            }
        }
        log.info("发送完成");
    }

}
